package com.supertool.tong.merger;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;
import java.util.Scanner;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapred.JobConf;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.codehaus.jackson.map.DeserializationConfig;
import org.codehaus.jackson.map.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.supertool.tong.merger.util.MZSequenceOutputFormat;
import com.supertool.tong.merger.util.MergeInputFormat;
import com.supertool.tong.merger.util.MergeOutputFormat;
import com.supertool.monitor.*;
import org.apache.hadoop.io.compress.SnappyCodec;

public class TongMerger extends Configured implements Tool{
    private static Logger LOG = LoggerFactory.getLogger(TongMerger.class);

    public static class MergeMapper extends Mapper<Comparable, Text, Comparable, Text> {
        public void map(Comparable key, Text value, Context context)
                throws IOException, InterruptedException {
            /*
			if (key != null && value != null) {
				context.write(key, value);
			}
			*/
        }
    }
    private void printUsageJson(){
        System.out.println("ZMerger {\"input_path\":[\"input_path1\",...],\"output_path\":\"output_path\",\"metafile\":\"metafile_path\"}");
    }

    protected void printUsagePlain(){
        System.out.println("ZMerger [-meta MetaFile] InputPattern [InputPattern ...] OutputPath");
    }

    private static class MergeArgs{
        public ArrayList<String> inputList;
        public String outputPath;
        public String metaFile;
        public String date;

        public List<String> getInput_path() {
            return inputList;
        }

        public void setInput_path(ArrayList<String> inputList) {
            this.inputList = inputList;
        }

        public String getOutput_path() {
            return outputPath;
        }
        public void setOutput_path(String outputPath) {
            this.outputPath = outputPath;
        }
        public String getMetafile() {
            return metaFile;
        }
        public void setMetafile(String metaFile) {
            this.metaFile = metaFile;
        }
        public void setDate( String date){
            this.date=date;
        }
        public String getDate( ){
            return date;
        }
    }

    private int errorLog(String info) {
        LOG.error(info);
        printUsageJson();
        return 1;
    }

    public int run(String[] args) throws Exception {
        if(args.length != 1){
            printUsageJson();
            return 1;
        }
        MergeArgs mergeArgs=null;
        try{
            Path confPath=new Path(args[0]);
            FileSystem fileSystem = confPath.getFileSystem(new Configuration());
            String content = new Scanner(fileSystem.open(confPath)).useDelimiter("\\Z").next();
            ObjectMapper mapper = new ObjectMapper();
            mapper.configure(DeserializationConfig.Feature.FAIL_ON_UNKNOWN_PROPERTIES, true);
            mergeArgs = mapper.readValue(content, MergeArgs.class);
        }catch (Exception e){
            return errorLog("Json error");
        }

        Configuration conf = new Configuration();
        conf.set("tong.parser.class", "com.supertool.monitor.LogParser");

        String date = "";
        if (mergeArgs.getDate().length() != 0) {
            date = mergeArgs.getDate();
        }

        if (mergeArgs.getMetafile().length() != 0) {
            LOG.info("meta found");
            conf.set("tong.meta.file", mergeArgs.getMetafile());
        } else {
            return errorLog("Empty metafile");
        }

        Job job = new Job(conf, "TONG_Merger" + date);
        job.setJarByClass(TongMerger.class);
        job.setMapperClass(MergeMapper.class);
        job.setNumReduceTasks(0);
        job.setInputFormatClass(MergeInputFormat.class);
        job.setOutputFormatClass(MZSequenceOutputFormat.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        if (mergeArgs.getInput_path().size() != 0) {
            for(String input: mergeArgs.getInput_path()) {
                MergeInputFormat.addInputPaths(job, input + "/*");
            }
        } else {
            return errorLog("Empty input_path");
        }

        if (mergeArgs.getOutput_path().length() != 0) {
            Path output=new Path(mergeArgs.getOutput_path());
            FileSystem fs=output.getFileSystem(conf);
            if (fs.exists(output)){
                Path tmpOutput=new Path(output.getParent(),"."+output.getName()+'.'+System.currentTimeMillis()/1000);
                fs.rename(output, tmpOutput);
            }
            FileOutputFormat.setOutputPath(job, output);
        } else {
            return errorLog("Empty output_path");
        }

        //FileInputFormat.setCompressOutput(job, true);
        FileOutputFormat.setCompressOutput(job, true);
        //FileOutputFormat.setOutputCompressorClass(job, SnappyCodec.class);

        job.submit();
        job.waitForCompletion(true);
        return 0;
        /*
        RunningJob rjob = JobClient.runJob(new JobConf(job.getConfiguration()));
        if (rjob.isSuccessful()) {
            LOG.info(("Job Done:" + job.getJobID()));
            return 0;
        } else {
            LOG.info(("Job ERROR:" + job.getJobID()));
            return 1;
        }
        */
    }

    public static void main(String[] args) throws Exception {
        int ret = ToolRunner.run(new TongMerger(), args);
        System.exit(ret);
    }
}
